/*
 * Copyright [2018] [Alex/libo(liboware@gmail.com)]
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *  http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package com.hyts.reactor.handler.reactor;

import java.util.function.Consumer;

import com.hyts.reactor.base.ProduceManager;
import com.hyts.reactor.handler.subcribe.Subscriber;
import com.hyts.reactor.handler.subcribe.Subscription;
import com.hyts.reactor.handler.subcribe.LambdaSubscriber;

/**
 * @author LiBo/Alex
 * @see ProduceManager<T>
 * @since V1.0
 * @version V1.0
 * Reactor stream model data stream deal manager
 */
public abstract class ReactorProducer<T>  {

    /**
     * register method
     * <b>bind the subscriber and productor relationship</b>
     * @param subscriber subcriber to consumer code or caller
     * @exception
     */
    protected abstract void register(Subscriber<? super T> subscriber);

    /**
     * consumer function
     * @param consumerFunction
     * @exception
     */
    protected void bind(Consumer<? super T> consumerFunction) {
        register(new LambdaSubscriber <>(consumerFunction,null,null,null));
    }

    /**
     * consumer function
     * <b>consumerFunction</b>
     * @param consumerFunction
     * @param errorConsumer
     * @exception
     */
    protected void bind(Consumer<T> consumerFunction,Consumer<? super Throwable> errorConsumer) {
        register(new LambdaSubscriber<>(consumerFunction,errorConsumer,null,null));
    }

    /**
     * consumer function
     * <b>consumerFunction/b>
     * @param consumerFunction
     * @param errorConsumer
     * @param completeConsumer
     * @exception
     */
    protected void bind(Consumer<T> consumerFunction,Consumer<? super Throwable> errorConsumer,Runnable completeConsumer) {
        register(new LambdaSubscriber<>(consumerFunction,errorConsumer,completeConsumer,null));
    }

    /**
     * consumer function
     * <b>consumerFunction</b>
     * @param consumerFunction
     * @param errorConsumer
     * @param completeConsumer
     * @param subscriptionConsumer
     * @exception
     */
    protected void bind(Consumer<T> consumerFunction,Consumer<? super Throwable> errorConsumer,Runnable completeConsumer,Consumer<? super Subscription> subscriptionConsumer) {
        register(new LambdaSubscriber<>(consumerFunction,errorConsumer,completeConsumer,subscriptionConsumer));
    }


    /**
     *
     * @return
     */
    public static <T> ReactorProducer<T> just(T... data){
        return new ReactorArrayProductor<T>(data);
    }

}
